package com.vodafone.mapreduce.reduce;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.atomic.AtomicBoolean;

import org.apache.log4j.Logger;

import com.vodafone.mapreduce.MapReduce;

/**
 * TODO: Insert your comments here!
 */
public abstract class ReduceFunction<L, W> {

    public static Logger log = MapReduce.LOG;
    protected final List<Map<L, List<W>>> combineResults;
    protected final Map<L, List<W>> intermediate = new HashMap<L, List<W>>();
    protected final AtomicBoolean ready = new AtomicBoolean(false);

    public ReduceFunction(List<Map<L, List<W>>> combineResults) {
        this.combineResults = combineResults;
    }

    public void reduce() {
        for (Map<L, List<W>> combineResult : combineResults) {
            for (Map.Entry<L, List<W>> entry : combineResult.entrySet()) {
                List<W> list = null;
                if (intermediate.containsKey(entry.getKey())) {
                    // Map entry in the combine intermediate already exists. Fetch it beforehand to make it available for reduction.
                    list = intermediate.get(entry.getKey());
                } else {
                    // Prepare a new list for reduction in case there is not yet an entry in the intermediate.
                    list = new ArrayList<W>();
                }
                // Reduce the new list and the existing list.
                doReduce(entry.getKey(), entry.getValue(), list);
                // Store back the result that now resides in the existing list.
                intermediate.put(entry.getKey(), list);
            }
        }
        ready.set(true);
    }

    protected abstract void doReduce(L l, List<W> newWs, List<W> oldWs);

    public Map<L, List<W>> getReduceResult() {
        if (ready.get()) {
            return intermediate;
        } else {
            return null;
        }
    }

}
